/*
 * Licensed under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.submarine.transport.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.submarine.common.Constants;
import org.submarine.common.Logger;
import org.submarine.common.URL;
import org.submarine.exchange.Request;
import org.submarine.exchange.Response;
import org.submarine.exchange.TransferObject;
import org.submarine.exchange.adapter.ExchangeAdapter;
import org.submarine.transport.Endpoint;
import org.submarine.transport.codec.NettyDecoder;
import org.submarine.transport.codec.NettyEncoder;

/**
 * 
 * @file org.submarine.transport.netty.NettyClient.java
 *
 * Netty客户端
 * 
 * @author gaojian
 * @created 2013-8-17 下午11:10:58
 * @version 0.0.1
 * @since 0.0.1
 *
 */
public class NettyClient extends Endpoint {
	
	private Bootstrap bootstrap;
	
	private EventLoopGroup group;
	
	private ConcurrentHashMap<URL, ChannelFuture> channelMap = new ConcurrentHashMap<URL, ChannelFuture>();
	
	protected final ConcurrentHashMap<Integer, Response> responseMap = new ConcurrentHashMap<Integer, Response>(256);
	
	private final AtomicInteger connectTimes = new AtomicInteger(0);
	
	private Lock lockChannelMap = new ReentrantLock();
	
	private Logger logger = Logger.getLogger(NettyClient.class);
	
	/**
	 * 构造器，启动
	 */
	public NettyClient(ExchangeAdapter exchangeAdapter) {
		super(exchangeAdapter);
		// Configure the client.
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
         .handler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) throws Exception {
                 ch.pipeline().addLast("decoder", new NettyDecoder(Constants.DEFAULT_CHARSET_NAME))
                 	.addLast("encoder", new NettyEncoder(Constants.DEFAULT_CHARSET_NAME))
                 	.addLast(
                 			new NettyClientHandler());
             }
         });
	}
	
	/* (non-Javadoc)
	 * @see org.submarine.startup.Endpoint#start()
	 */
	@Override
	public void start() {}
	
	/* (non-Javadoc)
	 * @see org.submarine.startup.Endpoint#stop()
	 */
	@Override
	public void stop() {
        group.shutdownGracefully();
	}
	
	public Response invokeSync(/*String address*/URL url, Request request, long timeoutMillis) throws InterruptedException {
		final Response response = new Response();
		Channel channel = this.getOrCreateChannel(url);
		if (null != channel && channel.isActive()) {
//			request.getData()
			channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if (future.isSuccess()) {
						response.setSendStatus(true);
						return;
					} else {
						response.setSendStatus(false);
					}
					response.setCause(future.cause());
				}
			});
			
			if (response.waitResponse(timeoutMillis)) {
				if (response.isSendStatus()) {
					throw new RuntimeException("Waiting response on channel <" + url.getAddress() + "> timeout( " + timeoutMillis + "(ms))", response.getCause());
				}  else {
					throw new RuntimeException("Send request to <" + url.getAddress() + "> failed", response.getCause());
				}
			}
		}
		return response;
	}
	
	public void invokeAsync() {
		
	}
	
	private Channel getOrCreateChannel(/*String address*/URL url) throws InterruptedException {
		ChannelFuture channelFuture = this.channelMap.get(url);
		if (null != channelFuture && channelFuture.channel().isActive()) {
			return channelFuture.channel();
		}
		
		if (this.lockChannelMap.tryLock(3000, TimeUnit.MILLISECONDS)) {
			try {
				boolean createConn = false;
				channelFuture = this.channelMap.get(url);
				if (null != channelFuture.channel() && channelFuture.channel().isActive()) {
					return channelFuture.channel();
				} else if (!channelFuture.isDone()) {
					createConn = true;
				} else {
					this.channelMap.remove(url);
					createConn = true;
				}
				if (createConn) {
//					while (null == channelFuture && connectTimes.getAndIncrement() < 3) {//重试3次
						channelFuture = this.bootstrap.connect(url.getHost(), url.getPort());
//					}
					connectTimes.set(0);
					this.channelMap.put(url, channelFuture);
				}
			} catch (Exception e) {
				logger.error("Create channel exception.", e);
			} finally {
				this.lockChannelMap.unlock();
			}
		}
		
		if (null != channelFuture) {
			if (channelFuture.awaitUninterruptibly(6000, TimeUnit.MILLISECONDS)) {
				if (null != channelFuture.channel() && channelFuture.channel().isActive()) {
					return channelFuture.channel();
				}
			}
		}
		
		return null;
	}
	
	class NettyClientHandler extends SimpleChannelInboundHandler<TransferObject> {

		@Override
		protected void channelRead0(ChannelHandlerContext ctx, TransferObject transferObject) throws Exception {
			processedMessageRecived(ctx, transferObject);
		}
		
	}
	
}
